package com.leigq.www.producer.sender;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.leigq.www.common.constant.QueueConstant;
import com.leigq.www.common.entity.Order;
import com.leigq.www.common.enums.KafKaMsgBizType;
import com.leigq.www.producer.entity.BrokerMessageLog;
import com.leigq.www.producer.enumeration.MessageLogStatusEnum;
import com.leigq.www.producer.service.BrokerMessageLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.nio.charset.StandardCharsets;

import static com.leigq.www.common.constant.QueueConstant.BZI_TYPE_RECORD_HEADER_KEY;

/**
 * KafKa订单发送
 * <br/>
 * 参考：<a href='https://blog.csdn.net/wangpf2011/article/details/86600809'>分布式架构核心组件之消息队列KafKa</a>
 * <p>
 * 创建人：asus <br>
 * 创建时间：2019-02-16 15:04 <br>
 * <p>
 * 修改人： <br>
 * 修改时间： <br>
 * 修改备注： <br>
 * </p>
 * @author leiguoqing
 */
@Component
@Slf4j
public class OrderSender implements KafKaSender<Order> {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * The Message log service.
     */
    @Autowired
    private BrokerMessageLogService messageLogService;

    /**
     * The Object mapper.
     */
    @Autowired
    private ObjectMapper objectMapper;

    /**
     * 发送消息方法调用: 构建自定义对象消息
     *
     * @param order the order
     */
    @Override
    public void send(Order order) {
        try {
            final String orderJson = objectMapper.writeValueAsString(order);
            // 设置消息唯一ID，这里从 订单 中获取消息唯一ID
            ProducerRecord<String, String> record = new ProducerRecord<>(QueueConstant.ORDER_TOPIC_NAME, order.getMsgId(), orderJson);
            final Headers headers = record.headers();
            // 自定义业务类型消息头
            headers.add(BZI_TYPE_RECORD_HEADER_KEY, (getBizType().code + "").getBytes(StandardCharsets.UTF_8));
            kafkaTemplate.send(record)
                    .addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
                        @Override
                        public void onFailure(Throwable ex) {
                            // 失败则进行具体的后续操作:重试 或者补偿等手段
                            System.out.println("发送消息异常");
                        }

                        @Override
                        public void onSuccess(SendResult<String, String> result) {
                            System.out.println("发送消息成功：" + orderJson);
                            // 消息发送成功
                            // 更新消息状态为成功：1
                            final ProducerRecord<String, String> producerRecord = result.getProducerRecord();
                            BrokerMessageLog log = new BrokerMessageLog();
                            log.setStatus(MessageLogStatusEnum.SUCCESS.getValue());
                            log.setMsgId(producerRecord.key());
                            messageLogService.update(log);
                        }
                    });
        } catch (JsonProcessingException e) {
            log.error("json转化异常:", e);
        }
    }

    @Override
    public KafKaMsgBizType getBizType() {
        return KafKaMsgBizType.ORDER_BIZ;
    }
}
